// $Id: queue.c 4 2011-10-20 16:07:10Z cedric.shih@gmail.com $
/*
 * Copyright 2011 Cedric Shih (cedric dot shih at gmail dot com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <stdlib.h>
#include <errno.h>
#include <string.h>

#include "log.h"

#include "queue.h"

void zcqueue_head_clear(struct zcqueue_head *list);

struct zcqueue *zcqueue_new(void)
{
	struct zcqueue *queue;

	zcbuffer_debug("constructing");

	if (!(queue = calloc(1, sizeof(*queue)))) {
		zcbuffer_error("colloc: %s", strerror(errno));
		return NULL;
	}

	TAILQ_INIT(&queue->caches);
	TAILQ_INIT(&queue->head);

	return queue;
}

void zcqueue_free(struct zcqueue *queue)
{
	if (!queue) {
		return;
	}

	zcbuffer_debug("destructing");

	zcqueue_head_clear(&queue->head);
	zcqueue_head_clear(&queue->caches);

	free(queue);
}

void zcqueue_head_clear(struct zcqueue_head *list)
{
	struct zcqueue_item *item;

	while ((item = TAILQ_FIRST(list))) {
		TAILQ_REMOVE(list, item, entry);
		free(item);
	}
}

size_t zcqueue_size(struct zcqueue *queue)
{
	if (!queue) {
		zcbuffer_error("null pointer");
		return 0;
	}
	return queue->size;
}

int zcqueue_put(struct zcqueue *queue,
		const struct timeval *timestamp, struct zcbuffer *buffer)
{
	int rc;
	struct zcqueue_item *item;

	if (!queue || !buffer) {
		rc = EINVAL;
		zcbuffer_error("null pointer(s)");
		goto finally;
	}

	zcbuffer_trace("putting to %p: %p", queue, buffer);

	if ((item = TAILQ_FIRST(&queue->caches))) {
		TAILQ_REMOVE(&queue->caches, item, entry);
	} else {
		if (!(item = calloc(1, sizeof(*item)))) {
			rc = errno ? errno : -1;
			zcbuffer_error("colloc: %s", strerror(rc));
			goto finally;
		}
	}

	if (timestamp) {
		memcpy(&item->timestamp, timestamp, sizeof(item->timestamp));
	} else {
		memset(&item->timestamp, '\0', sizeof(item->timestamp));
	}

	item->buffer = buffer;
	TAILQ_INSERT_TAIL(&queue->head, item, entry);
	queue->size++;
	rc = 0;

finally:
	return rc;
}

struct zcbuffer *zcqueue_peek(struct zcqueue *queue,
		struct timeval *timestamp)
{
	int rc;
	struct zcqueue_item *item;
	struct zcbuffer *buffer = NULL;

	if (!queue) {
		rc = EINVAL;
		zcbuffer_error("null pointer(s)");
		goto finally;
	}

	if (!(item = TAILQ_FIRST(&queue->head))) {
		rc = ENOENT;
		zcbuffer_error("nothing in queue");
		goto finally;
	}

	zcbuffer_trace("peeking from queue %p: %p", queue, item);

	if (timestamp) {
		memcpy(timestamp, &item->timestamp, sizeof(*timestamp));
	}

	buffer = item->buffer;
	rc = 0;

finally:
	if (rc != 0) {
		errno = rc;
	}
	return buffer;
}

struct zcbuffer *zcqueue_take(struct zcqueue *queue,
		struct timeval *timestamp)
{
	int rc;
	struct zcbuffer *buffer;
	struct zcqueue_item *item;

	if (!(buffer = zcqueue_peek(queue, timestamp))) {
		goto finally;
	}

	item = TAILQ_FIRST(&queue->head);

	zcbuffer_trace("taking from queue %p: %p", queue, item);

	TAILQ_REMOVE(&queue->head, item, entry);
	TAILQ_INSERT_TAIL(&queue->caches, item, entry);
	queue->size--;
	rc = 0;

finally:
	if (rc != 0) {
		errno = rc;
	}
	return buffer;
}

static inline int timeval_cmp(const struct timeval *t1,
		const struct timeval *t2)
{
	int result = t2->tv_sec - t1->tv_sec;

	if (!result) {
		result = t2->tv_usec - t1->tv_usec;
	}

	return result;
}

int zcqueue_take_newer(struct zcqueue *queue,
		struct zcqueue *dest, const struct timeval *timestamp)
{
	int rc;
	struct zcqueue_item *item, *next;

	if (!queue) {
		rc = EINVAL;
		zcbuffer_error("null pointer");
		goto finally;
	}

	zcbuffer_trace("taking items newer than %ld.%06ld in: %p",
			timestamp->tv_sec, timestamp->tv_usec, queue);

	TAILQ_FOREACH_SAFE(item, &queue->head, entry, next) {
		if (timeval_cmp(&item->timestamp, timestamp) >= 0) {
			continue;
		}
		TAILQ_REMOVE(&queue->head, item, entry);
		queue->size--;
		TAILQ_INSERT_TAIL(&dest->head, item, entry);
		dest->size++;
	}

	rc = 0;

finally:
	return rc;
}

int zcqueue_take_older(struct zcqueue *queue,
		struct zcqueue *dest, const struct timeval *timestamp)
{
	int rc;
	struct zcqueue_item *item, *next;

	if (!queue) {
		rc = EINVAL;
		zcbuffer_error("null pointer");
		goto finally;
	}

	zcbuffer_trace("taking items older than %ld.%06ld in: %p",
			timestamp->tv_sec, timestamp->tv_usec, queue);

	TAILQ_FOREACH_SAFE(item, &queue->head, entry, next) {
		if (timeval_cmp(&item->timestamp, timestamp) <= 0) {
			continue;
		}
		TAILQ_REMOVE(&queue->head, item, entry);
		queue->size--;
		TAILQ_INSERT_TAIL(&dest->head, item, entry);
		dest->size++;
	}

	rc = 0;

finally:
	return rc;
}

int zcqueue_clear(struct zcqueue *queue)
{
	int rc;
	struct zcqueue_item *item;

	if (!queue) {
		rc = EINVAL;
		zcbuffer_error("null pointer");
		goto finally;
	}

	zcbuffer_trace("clearing %p", queue);

	while ((item = TAILQ_FIRST(&queue->head))) {
		TAILQ_REMOVE(&queue->head, item, entry);
		TAILQ_INSERT_TAIL(&queue->caches, item, entry);
	}

	queue->size = 0;
	rc = 0;

finally:
	return rc;
}
